package com.sf.fy.uav.applet.config;

import com.sf.kafka.api.consume.*;
import com.sf.kafka.exception.KafkaException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

/**
 * @author LTQ
 */
@Component
public class KafkaConsumer {
    private boolean autoStartup = true;

    private ConsumeConfig consumeConfig;

    private static final Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);


    public KafkaConsumer() {
        super();

    }

    public KafkaConsumer(String systemName, String monitorUrl, String clusterName, String topic, String checkCode, int messageConsumeThreadCount) throws KafkaException {
        if (autoStartup) {
            Assert.notNull(systemName, "systemName can not be null!");
            Assert.notNull(checkCode, "checkCode can not be null!");
            Assert.notNull(monitorUrl, "monitorUrl can not be null!");
            Assert.notNull(clusterName, "clusterName can not be null!");
            Assert.notNull(topic, "topic can not be null!");
            final String systemIdToken = systemName + ":" + checkCode;
            consumeConfig = new ConsumeConfig(systemIdToken, monitorUrl, clusterName, topic, messageConsumeThreadCount);
        }
    }

    @SuppressWarnings({"unchecked", "rawtypes"})
    public void initRegister(final IMessageConsumeListener<?> messageConsumeListener) throws KafkaException {
        Assert.notNull(messageConsumeListener, "messageConsumeListener can not be null!");
        //开启线程,启动zookeeper注册
        Thread schedulerThread = new Thread() {
            @Override
            public void run() {
                try {
                    if (messageConsumeListener instanceof IStringMessageConsumeListener) {
                        KafkaConsumerRegister.registerStringConsumer(consumeConfig, (IStringMessageConsumeListener) messageConsumeListener);
                    } else if (messageConsumeListener instanceof IByteArrayMessageConsumeListener) {
                        KafkaConsumerRegister.registerByteArrayConsumer(consumeConfig, (IByteArrayMessageConsumeListener) messageConsumeListener);
                    } else {
                        logger.error("Unknown listener type!");
                    }
                } catch (Exception e) {
                    logger.error("kafka consumer service register exception:", e);
                }
            }

            ;
        };
        schedulerThread.setDaemon(true);    //设置为守护线程
        schedulerThread.start();
    }

}
